from kafka import KafkaConsumer
from kafka.errors import NoBrokersAvailable
import json
import os
import logging
import time

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

topic_name = 'local-test-topic'

def get_kafka_broker():
    return 'localhost:30094'

def create_consumer_with_retry(max_retries=10, retry_interval=10):
    """创建 consumer 并重试连接"""
    for attempt in range(max_retries):
        try:
            consumer = KafkaConsumer(
                bootstrap_servers=[get_kafka_broker()],
                auto_offset_reset='earliest',
                enable_auto_commit=True,
                group_id='kafka-demo-group',
                value_deserializer=lambda x: json.loads(x.decode('utf-8')) if x else None,
                key_deserializer=lambda x: x.decode('utf-8') if x else None,
                consumer_timeout_ms=10000,
                api_version=(2, 6, 0)  # 明确指定 API 版本
            )
            
            # 订阅主题
            consumer.subscribe([topic_name])
            logger.info(f"成功连接到 Kafka Broker: {get_kafka_broker()}")
            return consumer
            
        except NoBrokersAvailable as e:
            logger.warning(f"尝试 {attempt + 1}/{max_retries}: 无法连接到 Kafka, {retry_interval} 秒后重试...")
            if attempt == max_retries - 1:
                raise e
            time.sleep(retry_interval)
        except Exception as e:
            logger.warning(f"尝试 {attempt + 1}/{max_retries}: 连接异常, {retry_interval} 秒后重试... 错误: {e}")
            if attempt == max_retries - 1:
                raise e
            time.sleep(retry_interval)

def main():
    logger.info("启动 Kafka Consumer...")
    
    try:
        consumer = create_consumer_with_retry()
        
        logger.info("等待消息... (按 Ctrl+C 退出)")
        
        try:
            for message in consumer:
                if message.value:
                    logger.info("\n收到新消息:")
                    logger.info(f"  Topic: {message.topic}")
                    logger.info(f"  Partition: {message.partition}")
                    logger.info(f"  Offset: {message.offset}")
                    logger.info(f"  Key: {message.key}")
                    logger.info(f"  Value: {message.value}")
                    logger.info("-" * 50)
                else:
                    logger.warning("收到空消息")
                    
        except KeyboardInterrupt:
            logger.info("\n停止消费者...")
        except Exception as e:
            logger.error(f"消费消息时出错: {e}")
        finally:
            consumer.close()
            logger.info("Consumer 已关闭")
            
    except Exception as e:
        logger.error(f"Consumer 启动失败: {e}")

if __name__ == "__main__":
    main()
